package com.fwzhang.third.collectionframework.collection.queue;

import sun.misc.Unsafe;

import java.io.Serializable;
import java.util.AbstractQueue;
import java.util.Iterator;
import java.util.Queue;

/**
 * ConcurrentLinkedQueue
 * 基于链表的无界线程安全队列
 * 通过循环CAS实现线程安全, 非阻塞
 *
 * 1:不允许null入列
 * 2:在入队的最后一个元素的next为null
 * 3:队列中所有未删除的节点的item都不能为null且都能从head节点遍历到
 * 4:删除节点是将item设置为null, 队列迭代时跳过item为null节点
 * 5:head节点跟tail不一定指向头节点或尾节点，可能存在滞后性
 * https://www.jianshu.com/p/231caf90f30b
 * @param <E>
 */
public class ConcurrentLinkedQueueClass<E> extends AbstractQueue<E> implements Queue<E>, Serializable {

    /**
     * 无参构造函数
     * 创建头节点和尾节点, 节点内数据为null
     */
    public ConcurrentLinkedQueueClass() {
        head = tail = new Node<>(null);
    }

    /**
     * 队列节点类
     * @param <E>
     */
    private static class Node<E> {
        volatile E item;
        volatile Node<E> next;

        private static final Unsafe UNSAFE;
        private static final long itemOffset;
        private static final long nextOffset;

        static {
            try {
                UNSAFE = Unsafe.getUnsafe();
                Class<?> k = Node.class;
                itemOffset = UNSAFE.objectFieldOffset(k.getDeclaredField("item"));
                nextOffset = UNSAFE.objectFieldOffset(k.getDeclaredField("next"));
            } catch (Exception e) {
                throw new Error();
            }
        }

        Node(E item) {
            UNSAFE.putObject(this, itemOffset, item);
        }

        /**
         * 修改节点的值，cas修改
         * @param cmp
         * @param val
         * @return
         */
        boolean casItem(E cmp, E val) {
            return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
        }

        // 设置next节点
        void lazySetNext(Node<E> val) {
            UNSAFE.putOrderedObject(this, nextOffset, val);
        }

        /**
         * cas设置next节点
         * @param cmp
         * @param val
         * @return
         */
        boolean casNext(Node<E> cmp, Node<E> val) {
            return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
        }
    }

    // 队列头
    private transient volatile Node<E> head;
    // 队列尾
    private transient volatile Node<E> tail;


    /**
     * Throws NullPointerException if argument is null.
     *
     * @param v the element
     */
    private static void checkNotNull(Object v) {
        if (v == null)
            throw new NullPointerException();
    }

    private boolean casTail(Node<E> cmp, Node<E> val) {
        return UNSAFE.compareAndSwapObject(this, tailOffset, cmp, val);
    }

    private boolean casHead(Node<E> cmp, Node<E> val) {
        return UNSAFE.compareAndSwapObject(this, headOffset, cmp, val);
    }

    // Unsafe mechanics

    private static final sun.misc.Unsafe UNSAFE;
    private static final long headOffset;
    private static final long tailOffset;
    static {
        try {
            UNSAFE = sun.misc.Unsafe.getUnsafe();
            Class<?> k = ConcurrentLinkedQueueClass.class;
            headOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("head"));
            tailOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("tail"));
        } catch (Exception e) {
            throw new Error(e);
        }
    }

    final void updateHead(Node<E> h, Node<E> p) {
        if (h != p && casHead(h, p))
            h.lazySetNext(h);
    }

    /**
     * Returns an iterator over the elements contained in this collection.
     *
     * @return an iterator over the elements contained in this collection
     */
    @Override
    public Iterator<E> iterator() {
        return null;
    }

    /**
     * 遍历队列，但由于没有加锁，在遍历过程中如果出现offer,poll操作,会导致最终结果不准确
     * @return
     */
    @Override
    public int size() {
        return 0;
    }

    /**
     * 无界队列，不会返回false
     */
    @Override
    public boolean offer(E e) {

        checkNotNull(e);
        // 创建新的节点，final修饰，不可修改
        final Node<E> newNode = new Node<>(e);

        // 死循环
        /*
         * Node<E> t = tail;
         * Node<E> p = t;
         * for(;;)
         */
        for (Node<E> t = tail, p = t; ;) {
            Node<E> q = p.next;
            if (q == null) {
                // 找到队列当前的最后一个节点
                if (p.casNext(null, newNode)) {
                    if (p != t) {
                       casTail(t, newNode);
                    }
                    return true;
                }
            } else if (p == q) {
                // offer过程中出现了poll操作
                p = (t != (t = tail)) ? t : head;
            } else {
                // 获取当前尾节点
                p = (p != t && t != (t = tail)) ? t : q;
            }
        }
    }

    /**
     * Retrieves and removes the head of this queue,
     * or returns {@code null} if this queue is empty.
     *
     * @return the head of this queue, or {@code null} if this queue is empty
     */
    @Override
    public E poll() {
        restartFromHead:
        for (;;) {
            for (Node<E> h = head, p = h, q;;) {
                E item = p.item;

                if (item != null && p.casItem(item, null)) {
                    // Successful CAS is the linearization point
                    // for item to be removed from this queue.
                    if (p != h) // hop two nodes at a time
                        updateHead(h, ((q = p.next) != null) ? q : p);
                    return item;
                }
                else if ((q = p.next) == null) {
                    updateHead(h, p);
                    return null;
                }
                else if (p == q)
                    continue restartFromHead;
                else
                    p = q;
            }
        }
    }

    /**
     * Retrieves, but does not remove, the head of this queue,
     * or returns {@code null} if this queue is empty.
     *
     * @return the head of this queue, or {@code null} if this queue is empty
     */
    @Override
    public E peek() {
        restartFromHead:
        for (;;) {
            // h为当前线程的头节点，在当前线程中不变
            for (Node<E> h = head, p = h, q; ;) {
                E item = p.item;
                // item不等于null 或者当前线程的队列为空
                if (item != null || (q = p.next) == null) {
                    updateHead(h, p);
                    return item;
                } else if (p == q) {
                    // 其他线程修改了队列头，重新遍历
                    continue restartFromHead;
                } else {
                    p = q;
                }
            }
        }
    }
}
